package com.test.test5;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.math.BigDecimal;
import java.math.RoundingMode;

public class Reducer5 extends Reducer<LongWritable, Text, Text, Text> {
    final Text k = new Text();
    final Text v = new Text();
    // 昨日收盘价
    BigDecimal close = BigDecimal.ZERO;
    // 今日收盘价
    BigDecimal todayClose = BigDecimal.ZERO;
    // 每日收益率
    BigDecimal everyRatio = BigDecimal.ZERO;
    // 累积收益率
    BigDecimal multiplyRatio = BigDecimal.ZERO;

    @Override
    protected void reduce(LongWritable key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        for (Text value : values) {
            String line = value.toString();
            String[] split = line.split(",");
            String date = split[0];
            // 设置今日收盘价
            todayClose = new BigDecimal(split[1]);
            // 跳过第一天，因为第一天没有昨日收盘价
            if (key.get() == 1) {
                close = todayClose;
                break;
            }
            k.set(date + ",");
            // 计算每日收益率保留8位小数
            everyRatio = todayClose.subtract(close).divide(close, 8, RoundingMode.HALF_UP);
            // 更新昨日收盘价
            close = todayClose;
            // 计算累积收益率
            if (key.get() == 2) {
                multiplyRatio = BigDecimal.ONE.add(everyRatio).subtract(BigDecimal.ONE);
            } else {
                multiplyRatio = BigDecimal.ONE.add(multiplyRatio)
                        .multiply(BigDecimal.ONE.add(everyRatio))
                        .divide(BigDecimal.ONE, 8, RoundingMode.HALF_UP)
                        .subtract(BigDecimal.ONE);
            }
            v.set(multiplyRatio.toString());
            context.write(k, v);
        }
    }
}
